package _10_working_with_streaming_io

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl.Tcp.{IncomingConnection, ServerBinding}
import akka.stream.scaladsl._
import akka.util.ByteString

import scala.concurrent._

object EchoServer extends App {

  implicit val system       = ActorSystem("Tcp")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  val connections: Source[IncomingConnection, Future[ServerBinding]] =
    Tcp().bind("127.0.0.1", 8080)
  connections runForeach { connection =>
    println(s"New connection from: ${connection.remoteAddress }")

    val echo = Flow[ByteString]
               .via(Framing.delimiter(
                 ByteString("\n"),
                 maximumFrameLength = 256,
                 allowTruncation = true))
               .map(_.utf8String)
               .map(_ + "!!!\n")
               .map(ByteString(_))

    connection.handleWith(echo)
  }
}
